package com.jltx.xq.coroutinesandrxpro.rxjava

import android.os.Handler
import android.util.Log
import com.jltx.xq.coroutinesandrxpro.rxjava.RxJava2Learn1.TAG
import io.reactivex.Observable
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.functions.Consumer
import io.reactivex.schedulers.Schedulers
import java.util.concurrent.TimeUnit

/**
 * @author aidenquan
 *         Created by junlintianxia on 2021/07/26.
 */
object RxJava2Learn2 {

    fun intervalCancel(handle: Handler) {
        val disposabe = Observable.interval(1, TimeUnit.SECONDS).subscribe()

        handle.postDelayed({
            disposabe.dispose()
            Log.d(TAG, "Rx intervalCancel dispose${Thread.currentThread()}")

        }, 3000L)
    }

    fun customIntervalCancel(handle: Handler) {
        val disposabe =
            customInterval().subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .subscribe({ t -> Log.d(TAG, "Rx intervalCancel2 Next $t") }
                ) { t -> Log.d(TAG, "Rx intervalCancel2 Error $t") }

        handle.postDelayed({
            disposabe.dispose()
            Log.d(TAG, "Rx intervalCancel dispose${Thread.currentThread()}")

        }, 3000L)
    }

    private fun customInterval(): Observable<Int> = Observable.create { emitter ->
        for (i in 1..5) {
            if (!emitter.isDisposed) {
                emitter.onNext(i)
            } else {
                break
            }
        }
        emitter.onComplete()
    }
}